package com.lvlian;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.utils.OSUtils;
import com.utils.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.time.LocalDate;
import java.util.HashMap;
import java.util.Map;

public class LvlianSinkSubjectInfo extends RichSinkFunction<String> {
    private static final Logger LOG = LoggerFactory.getLogger(LvlianSinkSubjectInfo.class);
    private Connection connection;
    private PreparedStatement preparedStatementInsert;
    private PreparedStatement preparedStatementDelete;
    private String lastValue = null;
    private Map<Integer, KV> map = new HashMap<>();

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载驱动程序
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 获取数据库连接
        String url = "jdbc:mysql://192.168.0.26:3306/ry-cloud";
        String user = "root";
        String password = "gni2PbgxMNutdxz";
        if (OSUtils.isWindows()) {
            url = "jdbc:mysql://110.41.6.107:33306/ry-cloud";
            user = "root";
            password = "root";
        }
        connection = DriverManager.getConnection(url, user, password);
        // 创建 PreparedStatement 对象
        String sql = "replace INTO lv_subject_basic_info ( " +
                "id,sub_num,sub_type,sub_role,sub_name," +
                "credit_code,unit_type,registered_capital,legal_name,legal_id_type," +
                "legal_id_num,registered_date,expiration_date,business_scope,approved_date," +
                "gender,birthday,id_type,id_num,id_date," +
                "address,agency,registered_addr,contacts,phone," +
                "open_bank,bank_accno,rz_status,adcode,create_user," +
                "create_date,org_code"
                + ") VALUES (" +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?, ?, ?, ?, ?," +
                "?,?" +
                ")";

        LOG.info(sql);
        preparedStatementInsert = connection.prepareStatement(sql);
        map.put(1, new KV("ROW_ID", "id"));
        map.put(2, new KV("SUB_NUM", "sub_num"));
        map.put(3, new KV("SUB_TYPE", "sub_type"));
        map.put(4, new KV("SUB_ROLE", "sub_role"));
        map.put(5, new KV("SUB_NAME", "sub_name"));
        map.put(6, new KV("SC_CODE", "credit_code"));
        map.put(7, new KV("UNIT_TYPE", "unit_type"));
        map.put(8, new KV("REGISTERED_CAPITAL", "registered_capital"));
        map.put(9, new KV("LEGAL_NAME", "legal_name"));
        map.put(10, new KV("LEGAL_ID_TYPE", "legal_id_type"));
        map.put(11, new KV("LEGAL_ID_NUM", "legal_id_num"));
        map.put(12, new KV("REGISTERED_DATE", "registered_date"));
        map.put(13, new KV("EXPIRATION_DATE", "expiration_date"));
        map.put(14, new KV("BUSINESS_SCOPE", "business_scope"));
        map.put(15, new KV("APPROVED_DATE", "approved_date"));
        map.put(16, new KV("SEX", "gender"));
        map.put(17, new KV("BIRTHDAY", "birthday"));
        map.put(18, new KV("ID_TYPE", "id_type"));
        map.put(19, new KV("ID_NUM", "id_num"));
        map.put(20, new KV("ID_DATE", "id_date"));
        map.put(21, new KV("ADDRESS", "address"));
        map.put(22, new KV("AGENCY", "agency"));
        map.put(23, new KV("REGISTERED_ADDR", "registered_addr"));
        map.put(24, new KV("CONTACTS", "contacts"));
        map.put(25, new KV("TELEPHONE", "phone"));
        map.put(26, new KV("OPEN_BANK", "open_bank"));
        map.put(27, new KV("BANK_ACCNO", "bank_accno"));
        map.put(28, new KV("STS", "rz_status"));
        map.put(29, new KV("ADCODE", "adcode"));
        map.put(30, new KV("RECORDER", "create_user"));
        map.put(31, new KV("RECORD_DATE", "create_date"));
        map.put(32, new KV("ORG_CODE", "org_code"));
    }

    @Override
    public void invoke(String value, Context context) throws Exception {
        // 将 JSON 字符串解析为 JSONObject
        JSONObject jsonObject = JSON.parseObject(value);

        // 获取操作类型
        String operation = jsonObject.getString("op");

        // 判断操作类型
        if ("d".equals(operation)) {
            // 删除操作
            JSONObject beforeObject = jsonObject.getJSONObject("before");
            String id = beforeObject.getString("ROW_ID");
            preparedStatementDelete.setString(1, id);
            preparedStatementDelete.executeUpdate();
        } else {
            // 插入或更新操作
            JSONObject afterObject = jsonObject.getJSONObject("after");
            // 如果上一次处理的值为空，或者当前值与上一次处理的值不相同，则执行 sink 操作
            if (lastValue == null || !lastValue.equals(value)) {
                // 遍历原始表字段
                for (Map.Entry<Integer, KV> entry : map.entrySet()) {
                    String v = afterObject.getString(entry.getValue().getK());

                    LOG.info(entry.getKey() + ":-------------------------- " + entry.getValue().getK() + " -> " + entry.getValue().getV() + " -> " + v);
//                if ("PAY_TIME".equals(entry.getValue().getK()) || "RECORD_DATE".equals(entry.getValue().getK())) {
                    if ("REGISTERED_DATE".equals(entry.getValue().getK())
                            || "EXPIRATION_DATE".equals(entry.getValue().getK())
                            || "APPROVED_DATE".equals(entry.getValue().getK())
                            || "ID_DATE".equals(entry.getValue().getK())
                            || "BIRTHDAY".equals(entry.getValue().getK())
                            || "RECORD_DATE".equals(entry.getValue().getK())) {
                        if (v != null) {
                            v = StringUtils.restoreDate(v);
                        }


                    }
                    LOG.info(entry.getKey() + ": " + entry.getValue().getK() + " -> " + entry.getValue().getV() + " -> " + v);
                    preparedStatementInsert.setString(entry.getKey(), v);
                }

                preparedStatementInsert.executeUpdate();

                // 更新上一次处理的值为当前值
                lastValue = value;
            }
        }
    }


    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatementInsert != null) {
            preparedStatementInsert.close();
        }
        if (preparedStatementDelete != null) {
            preparedStatementDelete.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
